package org.huel.backend.websocket;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.*;
import org.huel.backend.config.WebSocketServer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @Kafka topic: LACount
 * @author Liweijian.
 * @Description 全体学生的L&A
 * @date 2024/11/16 17:34
 */
@Component
public class LACountConsumer {

    private static int totalL = 0;
    private static int totalA = 0;

    private KafkaConsumer<String, String> kfkConsumer;

    @PostConstruct
    public void init() {
        // 消费者配置信息
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "likai:9092");
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "LACountConsumer");
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 构建 Kafka 消费者
        kfkConsumer = new KafkaConsumer<>(properties);
        kfkConsumer.subscribe(Arrays.asList("LACount"));

        // 启动消费者线程
        new Thread(this::consumeMessages).start();
    }

    private void consumeMessages() {
        while (true) {
            // 拉取消息
            ConsumerRecords<String, String> conRecord = kfkConsumer.poll(Duration.ofMillis(2000));

            int currentL = 0;
            int currentA = 0;

            for (ConsumerRecord<String, String> consumer : conRecord) {
                String value = consumer.value();
                JSONObject jsonObject = JSONObject.parseObject(value);

                // 统计L&A
                if (jsonObject.getString("status").equals("L")) {
                    currentL += jsonObject.getIntValue("count");
                } else if (jsonObject.getString("status").equals("A")) {
                    currentA += jsonObject.getIntValue("count");
                }
            }

            // 更新总数
            totalL += currentL;
            totalA += currentA;

            // 创建输出的 JSON 对象
            JSONObject outputJson = new JSONObject();
            outputJson.put("totalL", totalL);
            outputJson.put("totalA", totalA);
            outputJson.put("currentL", currentL);
            outputJson.put("currentA", currentA);

            // 打印输出
//            System.out.println("================");
//            System.out.println(outputJson.toJSONString());
//            System.out.println("================");

            WebSocketServer.sendMessage("LACount", String.valueOf(outputJson));

        }
    }
}
